package com.metis.document.parse.dialog.service.message_listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.metis.document.parse.dialog.api.ParseControllerApi;
import com.metis.document.parse.dialog.constant.FileStatus;
import com.metis.document.parse.dialog.enums.FileTypeEnum;
import com.metis.document.parse.dialog.property.ParseTaskProperty;
import com.metis.document.parse.dialog.repo.TaskRepo;
import com.metis.document.parse.dialog.entity.ParseParameter;
import com.metis.document.parse.dialog.service.ParseService;
import com.metis.document.parse.dialog.utils.*;
import feign.FeignException;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.io.IOUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;

@Slf4j
@Component
public class ParseTaskMessageListener implements StreamListener<String, ObjectRecord<String, String>> {
    @Autowired
    private ParseControllerApi parseControllerApi;
    @Autowired
    @Lazy
    private RedisTemplate<String, String> redisStringTemplate;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private TaskRepo taskRepo;
    @Autowired
    private ThreadPoolUtils threadPoolUtils;
    @Autowired
    @Lazy
    private ParseTaskMessageListener self;
    @Autowired
    private ParseService parseService;
    private final ParseTaskProperty parseTaskProperty = new ParseTaskProperty();

    @Data
    public static class Message {
        private Long taskId;
        private String filename;
        private String filepath;
        private String resultFilepath;
        private Integer retry = 0;
        private Long filesize;
    }

    public static final String STREAM_KEY = "METIS:DOCUMENT:PARSE:DIALOG:TASK:STREAM";
    public static final String STREAM_GROUP = "PARSE_NODE";

    @Override
    @Async("parseTaskExecutor")
    public void onMessage(ObjectRecord<String, String> message) {
        Message messageContent;
        try {
            String value = message.getValue();
            messageContent = objectMapper.readerFor(Message.class).readValue(value);
            CompletableFuture<Void> future;
            //如果文件太大
            if (messageContent.getFilesize() > 5 * 1024 * 1024) {
                future = threadPoolUtils.heightThreadPool(() -> doMessageConsumer(messageContent,message));
            } else {
                future = threadPoolUtils.lowThreadPool(() -> doMessageConsumer(messageContent,message));
            }
            future.join();
        } catch (JsonProcessingException e) {
            log.error("parse message error, message:{}", message, e);
        } catch (Exception e) {
            log.error("unknown error", e);
        } finally {
            redisStringTemplate.opsForStream().acknowledge(STREAM_KEY,STREAM_GROUP, message.getId());
            redisStringTemplate.opsForStream().delete(STREAM_KEY,message.getId());
            log.info("acknowledge message:{}", message.getId());
//            System.gc();
        }
    }

    @Retryable(value = FeignException.FeignClientException.class, maxAttempts = 3, backoff = @Backoff( // 重试等待策略
            delay = 10 * 1000L,// 首次重试间隔10s
            multiplier = 2// 多次重试间隔系数2
    ))
    public void doMessageConsumer(@NotNull Message message,ObjectRecord<String,String> oms) {
        Long taskId = message.getTaskId();
        try {
            log.info("开始处理解析任务: {}", message);
            if (message.getRetry() > 0) {
                resetTask(message.getTaskId());
            }
            startParse(taskId);
            File taskFile = new File(message.getFilepath());
            //获取文件后缀
            AutoCleanFile temp = new AutoCleanFile(FileTypeEnum.ZIP);

            String contentType = Files.probeContentType(taskFile.toPath());
            try (FileOutputStream tos = new FileOutputStream(temp); FileInputStream tis = new FileInputStream(temp);
                 ByteArrayMultipartFile multipartFile = new ByteArrayMultipartFile(taskFile.getName(), taskFile.getName(), contentType);

                 MessageDigestInputStream mis = new MessageDigestInputStream(new FileInputStream(taskFile))) {
                IOUtils.copy(mis, multipartFile);

                String fileName = taskFile.getName();
                String fileExt = fileName.substring(fileName.lastIndexOf('.') + 1);
                ParseParameter parameter = new ParseParameter(multipartFile, tos, fileName, fileExt);

                //parseService.parsePDF(multipartFile, tos);
                parseService.parsePDF(parameter);
                String baseName = FilenameUtils.getBaseName(message.getFilename());
                File dstDir = new File(new File(message.getResultFilepath()).getParentFile(), baseName);
                FileUtils.deleteDirectory(dstDir);
                Files.createDirectories(dstDir.toPath());
                try (final ZipReader reader = new ZipReader(tis, StandardCharsets.UTF_8)) {
                    reader.readTo(dstDir);
                }
                //覆盖json
                reformatJson(taskId, dstDir.getParentFile().getName(), FilenameUtils.getBaseName(message.getFilename()), new File(dstDir, "output.json"));
                //移动文件
                Files.move(Path.of(dstDir.getAbsolutePath(), "output.md"), Path.of(dstDir.getAbsolutePath(), String.format("%s.md", baseName)));
                Files.move(Path.of(dstDir.getAbsolutePath(), "output.json"), Path.of(dstDir.getAbsolutePath(), String.format("%s.json", baseName)));
                taskSuccess(taskId);
                //redisStringTemplate.opsForStream().acknowledge(STREAM_GROUP, oms);
            }
        } catch (Throwable e) {
            log.error("parse error, taskId:{}", taskId, e.getMessage());
            //redisStringTemplate.opsForStream().acknowledge(STREAM_GROUP, oms);
            taskFail(taskId, e, message);
        }
    }

    private void startParse(@NotNull Long id) {
        taskRepo.findById(id).ifPresent(taskPO -> {
            taskPO.setStatus(FileStatus.PARSING);
            taskPO.setStart_time(LocalDateTime.now());
            taskRepo.save(taskPO);
        });
    }

    private void taskSuccess(@NotNull Long id) {
        taskRepo.findById(id).ifPresent(taskPO -> {
            taskPO.setStatus(FileStatus.PARSE_SUCCESS);
            taskPO.setEnd_time(LocalDateTime.now());
            taskRepo.save(taskPO);
        });
    }

    private void taskFail(Long id, Throwable e, Message message) {
        //重发消息进行重试
        message.setRetry(message.getRetry() + 1);
        String messageJson;
        try {
            messageJson = objectMapper.writeValueAsString(message);
        } catch (JsonProcessingException ex) {
            throw new RuntimeException(ex);
        }
        taskRepo.findById(id).ifPresent(taskPO -> {
            taskPO.setStatus(FileStatus.PARSE_FAIL);
            taskPO.setEnd_time(LocalDateTime.now());
            taskPO.setFail_reason(e.getMessage());
            taskRepo.save(taskPO);
        });
//        if (message.getRetry() < 4) {
//            redisStringTemplate.opsForStream().add(Record.of(messageJson).withStreamKey(ParseTaskMessageListener.STREAM_KEY));
//        }
    }

    private void taskSkip(Long id, String checksum) {
        taskRepo.findById(id).ifPresent(taskPO -> {
            taskPO.setStatus(FileStatus.PARSE_SKIP);
            taskPO.setChecksum(checksum);
            taskPO.setEnd_time(LocalDateTime.now());
            taskRepo.save(taskPO);
        });
    }

    private void resetTask(Long id) {
        taskRepo.findById(id).ifPresent(taskPO -> {
            taskPO.setStatus(FileStatus.WAITING);
            taskPO.setEnd_time(null);
            taskPO.setStart_time(null);
            taskPO.setFail_reason(null);
            taskPO.setChecksum(null);
            taskRepo.save(taskPO);
        });
    }

    private void reformatJson(long id, String dataType, String title, File file) throws Exception {
        if (parseTaskProperty.getDataTypeMap().containsKey(dataType)) {
            dataType = parseTaskProperty.getDataTypeMap().get(dataType);
        } else {
            log.warn("未知的数据类型:{}", dataType);
        }
        try (final InputStream is = new FileInputStream(file)) {
            JSONObject jsonObject = JSON.parseObject(is, JSONObject.class);
            //JSONObject jsonObject = new JSONObject();
            jsonObject.put("id", id);
            jsonObject.put("data_type", dataType);
            jsonObject.put("title", title);
            //jsonObject.put("content", section_content);
            FileUtils.writeStringToFile(file, jsonObject.toJSONString(), StandardCharsets.UTF_8);
        }
    }
}
